package cn.itcast.map;

import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import cn.itcast.util.DbUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.Date;
import java.util.Map;

//通用业务处理代码：日，周，月
public class StockKlineMap extends RichMapFunction<StockBean, Row> {

    private String kType;
    private String firstTxdate; //周期首个交易日字段名

    //  DAYK("1", "trade_date"),
    //	WEEKK("2", "week_first_txdate"),
    //	MONTHK("3", "month_first_txdate"),
    //	YEARK("4", "year_first_txdate");
    public StockKlineMap(String type, String firstTxDateType) {
        this.kType = type;
        this.firstTxdate = firstTxDateType;
    }

    String tradeDate = null;
    String firstTradeDate = null;
    //  secCode  每个内部map是一行数据,key是字段名，value是字段值
    Map<String, Map<String, Object>> klineMap=null;

    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * 一、初始化
         *      1.创建构造方法
         *        入参：kType：K线类型
         *             firstTxdate：周期首个交易日字段名
         *      2.获取交易日历表交易日数据
         *      3.获取周期首个交易日和T日
         *      4.获取K(周、月)线下的汇总表数据（高、低、成交量、金额）
         */
        //todo 2.获取交易日历表交易日数据
        String sql = "select * from tcc_date where trade_date<=curdate() order by trade_date desc limit 1";
        Map<String, String> dataMap = DbUtil.queryKv(sql); //列名为key键，列值为value
        //todo 3.获取周期首个交易日和T日
        //如果是日K，tradeDate=firstTradeDate
        //如果不是日K, tradeDate>=firstTradeDate（如果T日是周期首个交易日，则日/周/月都是同一天，否则大于）
        firstTradeDate = dataMap.get(firstTxdate);
        tradeDate = dataMap.get("trade_date");
        //todo 4.获取K(周、月)线下的汇总表数据（高、低、成交量、金额）
        String sqlKline = "select sec_code,\n" +
                "       MAX(high_price) as high_price,\n" +
                "       MIN(low_price) as low_price,\n" +
                "       SUM(trade_vol) as trade_vol,\n" +
                "       SUM(trade_amt) as trade_amt\n" +
                "from bdp_quot_stock_kline_day\n" +
                "where trade_date between "+firstTradeDate+" and "+tradeDate+" \n" +
                "group by 1";

        klineMap = DbUtil.query("sec_code", sql);
    }

    @Override
    public Row map(StockBean value) throws Exception {
        /**
         * 业务处理
         *      1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
         *      2.获取T日和周首次交易日时间,转换成long型
         *      3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
         *      4.获取周/月K数据：成交量、成交额、高、低
         *      5.高、低价格比较
         *      6.计算成交量、成交额
         *      7.计算均价
         *      8.封装数据Row
         */

        //todo 1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
        BigDecimal preClosePrice = value.getPreClosePrice();
        BigDecimal closePrice = value.getClosePrice();
        BigDecimal openPrice = value.getOpenPrice();
        BigDecimal highPrice = value.getHighPrice();
        BigDecimal lowPrice = value.getLowPrice();
        Long tradeVolDay = value.getTradeVolDay();
        Long tradeAmtDay = value.getTradeAmtDay();
        //todo 2.获取T日和周首次交易日时间,转换成long型,进行比较运算，但日历表的数据是(yyyy-mm-dd)类型
        Long tradeTime = DateUtil.stringToLong(tradeDate, Constant.format_yyyy_mm_dd);
        //周期内首个交易日
        Long firstTradeTime = DateUtil.stringToLong(firstTradeDate, Constant.format_yyyy_mm_dd);
        //todo 3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
        if(firstTradeTime<tradeTime && (kType.equals("2") || kType.equals("3"))){//一定是周K或月K

            //todo 4.获取周/月K数据：成交量、成交额、高、低
            Map<String, Object> map = klineMap.get(value.getSecCode());
            if(map!=null && map.size()>0){//说明有数据
                //历史总成交量
                Long tradeVol = Long.valueOf(map.get("trade_vol").toString());
                //历史总成交金额
                Long tradeAmt = Long.valueOf(map.get("trade_amt").toString());
                //todo 6.计算成交量、成交额
                tradeVolDay += tradeVol;//周期内的最新总成交量
                tradeAmtDay += tradeAmt;//周期内的最新总成交金额

                //获取历史最高价
                BigDecimal high_price = new BigDecimal(map.get("high_price").toString());
                BigDecimal low_price = new BigDecimal(map.get("low_price").toString());

                //todo 5.高、低价格比较获取最新的最高价
                if(high_price.compareTo(highPrice)==1) {//>
                    highPrice = high_price;
                }
                if(low_price.compareTo(lowPrice)==-1){ //<
                    lowPrice = low_price;
                }
            }
        }
        //todo 7.计算均价 （成交金额/成交量）
        BigDecimal avgPrice = new BigDecimal(0);
        if(tradeVolDay!=0){
            avgPrice = new BigDecimal(tradeAmtDay).divide(new BigDecimal(tradeVolDay),2, RoundingMode.HALF_UP);
        }
        //todo 8.封装数据Row
        Row row = new Row(13);
        /**
         *     update_time     timestamp       '更新时间',
         *     trade_date      date            '交易日期',
         *     sec_code        char(6)         '个股代码',
         *     sec_name        varchar(20)     '个股名称',
         *     k_line_type     char            'K线类型',
         *     pre_close_price decimal(8, 2)   '前收盘价',
         *     open_price      decimal(8, 2)   '开盘价',
         *     high_price      decimal(8, 2)   '最高价',
         *     low_price       decimal(8, 2)   '最低价',
         *     close_price     decimal(8, 2)   '收盘价',
         *     avgprice        decimal(8, 2)   '均价',
         *     trade_vol       decimal(18)     '成交量',
         *     trade_amt       decimal(18, 2)  '成交金额',
         */
        row.setField(0,new Timestamp(new Date().getTime()));
        row.setField(1,tradeDate);
        row.setField(2,value.getSecCode());
        row.setField(3,value.getSecName());
        row.setField(4,kType);
        row.setField(5,preClosePrice);
        row.setField(6,openPrice);
        row.setField(7,highPrice);
        row.setField(8,lowPrice);
        row.setField(9,closePrice);
        row.setField(10,avgPrice);
        row.setField(11,tradeVolDay);
        row.setField(12,tradeAmtDay);
        return row;
    }
}
